/*
 * manager.cc
 *
 *  Created on: Mar 21, 2012
 *      Author: yxjiang
 */

#include "manager.h"

namespace event
{

using namespace std;
using namespace boost::property_tree;

bool Manager::service_stop_ = false;
int Manager::state_port_ = 0;
float Manager::rate_ = 1.0;
map<string, CrawlerStatus> Manager::crawlers_;

Manager::Manager(int port, float rate, string collector_register_ip, int collector_register_port)
	: collector_register_ip_(collector_register_ip), collector_register_port_(collector_register_port)
{
	state_port_ = port;
	rate_ = rate;
}

Manager::~Manager()
{
	service_stop_ = true;
	//	wait the web service to stop
	if (web_service_pid_ > 0)
		pthread_join(web_service_pid_, NULL);

	//	wait all crawlers to stop
	map<string, CrawlerStatus>::iterator itr = crawlers_.begin();
	for (; itr != crawlers_.end(); ++itr)
	{
		if (itr->second.pid > 0)
			pthread_join(itr->second.pid, NULL);
	}
}

void Manager::SetRate(float rate)
{
	rate_ = rate;
}

void Manager::Attach(Crawler *crawler)
{
	CrawlerStatus status;
	status.crawler = crawler;
	status.running = false;
	status.pid = 0;
	crawlers_.insert(make_pair<string, CrawlerStatus>(crawler->GetCrawlerName(), status));
}

void Manager::Detach(const string &name)
{
	crawlers_.erase(name);
}

Crawler *Manager::GetCrawler(const string &name) const
{
	return crawlers_.find(name)->second.crawler;
}

/*
 * Start all the crawlers as separate threads,
 * register the monitoring module to the collector,
 * and finally start the web service to send the meta-data.
 */
void Manager::StartService()
{
	//	start all crawler threads
	map<string, CrawlerStatus>::iterator itr = crawlers_.begin();
	for (; itr != crawlers_.end(); ++itr)
	{
		pthread_create(&(itr->second.pid), NULL, _CrawlerService, (void*)(itr->second.crawler));
	}

	_SelfRegistration(collector_register_ip_, collector_register_port_);

	//	start web service
	pthread_create(&web_service_pid_, NULL, _WebService, NULL);
	itr = crawlers_.begin();
	for(; itr != crawlers_.end(); ++itr)
		if(itr->second.pid > 0)
			pthread_join(itr->second.pid, NULL);
}

void *Manager::_CrawlerService(void *arg)
{
	Crawler *crawler = (Crawler*)arg;
	while (service_stop_ == false){
		ThreadSleep(rate_, 0);
		crawler->FetchData();
	}
	return NULL;
}

/*
 * Provide the web service to subscriber who wants the data.
 */
void *Manager::_WebService(void *arg)
{
	struct sockaddr_in server_addr;	// Server Internet address
	//	initialize server address
	bzero(&server_addr, sizeof(server_addr));
	server_addr.sin_family = AF_INET;
	server_addr.sin_addr.s_addr = htons(INADDR_ANY);
	server_addr.sin_port = htons(state_port_);

	int server_sock = socket(AF_INET, SOCK_STREAM, 0);
	if(server_sock < 0)
	{
		fprintf(stderr, "Monitoring service creates socket failed.\n");
		exit(1);
	}
	else
		printf("Monitoring service socket created...\n");

	//	bind socket and address
	if(bind(server_sock, (struct sockaddr*)&server_addr, sizeof(server_addr)))
	{
		fprintf(stderr, "Monitoring service bind port: %d failed.\n", state_port_);
		exit(1);
	}
	else
		printf("Monitoring service port binded to port %d...\n", state_port_);

	//	listen
	if(listen(server_sock, 5))
	{
		fprintf(stderr, "Monitoring service listen failed.\n");
		exit(1);
	}
	else
		printf("Monitoring service listening on port %d...\n", state_port_);

	while(service_stop_ == false)
	{
		int conn_sock = accept(server_sock, NULL, 0);
		const string str_content = _AssembleJson();
		const char *content = str_content.c_str();
//		write(conn_sock, content, strlen(content));
		int nbytes;
		int left = strlen(content);
		int offset = 0;
//		send(conn_sock, content, strlen(content), 0);
		while((nbytes = send(conn_sock, content + offset, strlen(content + offset), 0)) != left)
		{
			offset += nbytes;
			left -= offset;
		}
		close(conn_sock);
	}

	return NULL;
}

const string Manager::_AssembleJson()
{
	ptree root;
	map<string, CrawlerStatus>::iterator crawler_itr = crawlers_.begin();
	//	put timestamp of the first crawler
	ObserveData cur_data = crawler_itr->second.crawler->GetData();
	root.put<long int>("timestamp", cur_data.timestamp);	//	set timestamp

	//	iterates all the crawlers to assemble the monitoring meta-data
	for(; crawler_itr != crawlers_.end(); ++crawler_itr)
	{
		string stream_name = crawler_itr->second.crawler->GetCrawlerName();
		ObserveData data = crawler_itr->second.crawler->GetData();
		ptree stream_node;
		boost::shared_ptr<map<string, string> > properties = data.properties_;
		map<string, string>::iterator property_itr = properties->begin();
		for(; property_itr != properties->end(); ++property_itr)
			stream_node.put<string>(property_itr->first, property_itr->second);

		root.add_child(stream_name, stream_node);
	}

	stringstream ss;
	write_json(ss, root);
	string strJson = ss.str();

	if(strJson.size() == 0)
	{
		fprintf(stderr, "size = 0\n");
	}

	return strJson;
}

/*
 * Self register the machine to the collector as a data stream.
 */
void Manager::_SelfRegistration(string collector_register_ip, int collector_register_port)
{
	//	register to the collector
	struct hostent *server_host = NULL;
	server_host = gethostbyname(collector_register_ip.c_str());
	struct sockaddr_in collector_addr;
	bzero((char*) &collector_addr, sizeof(collector_addr));
	collector_addr.sin_family = AF_INET;
	bcopy((char*) server_host->h_addr, (char*) &collector_addr.sin_addr.s_addr, server_host->h_length);
	collector_addr.sin_port = htons(collector_register_port);
	int collector_socket = socket(AF_INET, SOCK_STREAM, 0);

	if (connect(collector_socket, (struct sockaddr*) &collector_addr, sizeof(collector_addr)) < 0)
	{
		fprintf(stderr, "Cannot connect to collector!\n");
	}
	else
	{
		//	load network interface information
		struct ifaddrs *ifAddrStruct = NULL;
		struct ifaddrs *ifa = NULL;
		void *tmpAddrPtr = NULL;
		getifaddrs(&ifAddrStruct);
		char hostname[256];
		bzero(hostname, strlen(hostname));
		gethostname(hostname, 256);

		for (ifa = ifAddrStruct; ifa != NULL; ifa = ifa->ifa_next)
		{
			if (ifa->ifa_addr->sa_family == AF_INET) {
				// check it is IP4 is a valid IP4 Address
				tmpAddrPtr = &((struct sockaddr_in *) ifa->ifa_addr)->sin_addr;
				char addressBuffer[INET_ADDRSTRLEN];
				inet_ntop(AF_INET, tmpAddrPtr, addressBuffer, INET_ADDRSTRLEN);
				if (strcmp(ifa->ifa_name, "eth0") == 0
						|| strcmp(ifa->ifa_name, "wlan0") == 0)	//	find first wired or wireless network
				{
					//	register by sending ip, port, and name
					stringstream ss;
					ss << hostname << "\t" << addressBuffer << "\t" << state_port_;
					const char *content = ss.str().c_str();
					fprintf(stderr, "Register to collector by sending: %s.\n", content);
					if(send(collector_socket, content, strlen(content), 0) != strlen(content))
					{
						fprintf(stderr, "Self-registration failed.\n");
					}
					else
					{
						fprintf(stderr, "Self-registration successfully.\n");
					}
					break;
				}
			}
		}
		if (ifAddrStruct != NULL)
			freeifaddrs(ifAddrStruct);
	}
	close(collector_socket);
}

};

int main(int argc, char *argv[])
{
	using namespace event;
	using namespace std;

	int port = 32167;
	if(argc >= 2)
		port = atoi(argv[1]);

	fprintf(stderr, "Start to monitor...\n");

	CPUCrawler *crawler1 = new CPUCrawler("cpu", CPUCrawler::SEPARATE_CPU);
	MemCrawler *crawler2 = new MemCrawler("mem");
	NetCrawler *crawler3 = new NetCrawler("net");
	DiskCrawler *crawler4 = new DiskCrawler("disk");
	DiskIOCrawler *crawler5 = new DiskIOCrawler("diskio");

	float rate = 0.5;
	string registration_addr = "131.94.130.167";
	int registration_port = 32166;
	Manager manager(port, rate, registration_addr.c_str(), registration_port);
	manager.Attach(crawler1);
	manager.Attach(crawler2);
	manager.Attach(crawler3);
	manager.Attach(crawler4);
	manager.Attach(crawler5);

	manager.StartService();
	return 1;
}
